/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.client.impl.consumer;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerOrderly;
import com.alibaba.rocketmq.client.hook.ConsumeMessageContext;
import com.alibaba.rocketmq.client.log.ClientLogger;
import com.alibaba.rocketmq.client.stat.ConsumerStatsManager;
import com.alibaba.rocketmq.common.ThreadFactoryImpl;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.alibaba.rocketmq.common.protocol.body.CMResult;
import com.alibaba.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import com.alibaba.rocketmq.remoting.common.RemotingHelper;

/**
 *
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-6-27
 */
public class ConsumeMessageOrderlyService implements ConsumeMessageService {
	private static final Logger log = ClientLogger.getLog();
	private final static long MaxTimeConsumeContinuously = Long.parseLong(System.getProperty("rocketmq.client.maxTimeConsumeContinuously", "60000"));

	private volatile boolean stopped = false;

	private final DefaultMQPushConsumerImpl defaultMQPushConsumerImpl;
	private final DefaultMQPushConsumer defaultMQPushConsumer;
	private final MessageListenerOrderly messageListener;
	private final BlockingQueue<Runnable> consumeRequestQueue;
	private final ThreadPoolExecutor consumeExecutor;
	private final String consumerGroup;
	private final MessageQueueLock messageQueueLock = new MessageQueueLock();

	private final ScheduledExecutorService scheduledExecutorService;

	public ConsumeMessageOrderlyService(DefaultMQPushConsumerImpl defaultMQPushConsumerImpl, MessageListenerOrderly messageListener) {
		this.defaultMQPushConsumerImpl = defaultMQPushConsumerImpl;
		this.messageListener = messageListener;

		this.defaultMQPushConsumer = this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer();
		this.consumerGroup = this.defaultMQPushConsumer.getConsumerGroup();
		this.consumeRequestQueue = new LinkedBlockingQueue<Runnable>();

		this.consumeExecutor = new ThreadPoolExecutor(//
				this.defaultMQPushConsumer.getConsumeThreadMin(), //
				this.defaultMQPushConsumer.getConsumeThreadMax(), //
				1000 * 60, //
				TimeUnit.MILLISECONDS, //
				this.consumeRequestQueue, //
				new ThreadFactoryImpl("ConsumeMessageThread_"));

		this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("ConsumeMessageScheduledThread_"));
	}

	public void start() {
		// 顺序消费
		// 当消息模式为集群模式时，设置定时任务每隔20秒调用一次ConsumeMessageOrderlyService. lockMQPeriodically()方法获取消息队列的分布式锁
		if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())) {
			this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
				@Override
				public void run() {
					// 顺序消费的三把锁
					// 锁1====》首先在ConsumeMessageOrderlyService类中定义了定时任务每隔20秒执行一次lockMQPeriodically()方法，
					// 获取该Consumer端在Broker端锁住的MessageQueue集合（即分布式锁），
					// 并将RebalanceImpl.processQueueTable:ConcurrentHashMap<MessageQueue, ProcessQueue>集合中获得分布式锁的MessageQueue对象（消费队列）对应的ProcessQueue对象（消费处理队列）加上本地锁（即该对象的lock等于ture）以及加锁的时间，
					// 目的是为了在消费时在本地检查消费队列是否锁住；
					ConsumeMessageOrderlyService.this.lockMQPeriodically();
				}
			}, 1000 * 1, ProcessQueue.RebalanceLockInterval, TimeUnit.MILLISECONDS);
		}
	}

	public void shutdown() {
		this.stopped = true;
		this.scheduledExecutorService.shutdown();
		this.consumeExecutor.shutdown();
		if (MessageModel.CLUSTERING.equals(this.defaultMQPushConsumerImpl.messageModel())) {
			this.unlockAllMQ();
		}
	}

	public synchronized void unlockAllMQ() {
		this.defaultMQPushConsumerImpl.getRebalanceImpl().unlockAll(false);
	}

	public synchronized void lockMQPeriodically() {
		if (!this.stopped) {
			this.defaultMQPushConsumerImpl.getRebalanceImpl().lockAll();
		}
	}

	public synchronized boolean lockOneMQ(final MessageQueue mq) {
		if (!this.stopped) {
			return this.defaultMQPushConsumerImpl.getRebalanceImpl().lock(mq);
		}

		return false;
	}

	public void tryLockLaterAndReconsume(final MessageQueue mq, final ProcessQueue processQueue, final long delayMills) {
		this.scheduledExecutorService.schedule(new Runnable() {
			@Override
			public void run() {
				/**
				 * 1、先调用ConsumeMessageOrderlyService.lockOneMQ(MessageQueue mq)方法
				 * 获取MessageQueue队列的锁，
				 * 向该MessageQueue对象的brokerName下面的主用Broker发送LOCK_BATCH_MQ请求码的请求消息，
				 * 请求Broker将发送的MessageQueue对象锁住；
				 * 若该请求的MessageQueue对象在Broker返回的锁住集合中，则锁住成功了；
				 */
				boolean lockOK = ConsumeMessageOrderlyService.this.lockOneMQ(mq);

				// 2、调用ConsumeMessageOrderlyService.submitConsumeRequestLater(ProcessQueue processQueue, MessageQueue messageQueue, long suspendTimeMillis)方法，
				// 若锁住成功则suspendTimeMillis=10；
				// 若未锁住，则suspendTimeMillis=3000。
				if (lockOK) {
					// 3、在ConsumeMessageOrderlyService.submitConsumeRequestLater方法中，
					// 初始化一个匿名的Runnable线程类，在suspendTimeMillis毫秒之后，执行该线程类，
					// 在该类的run方法中调用ConsumeMessageOrderlyService.submitConsumeRequest(List<MessageExt> msgs, ProcessQueue processQueue, MessageQueue messageQueue, boolean dispathToConsume)方法，
					// 其中dispathToConsume=true；
					// 在该方法中根据ProcessQueue 和MessageQueue对象初始化ConsumeRequest对象，并放入ConsumeMessageOrderlyService.consumeExecutor线程池中；
					ConsumeMessageOrderlyService.this.submitConsumeRequestLater(processQueue, mq, 10);
				} else {
					ConsumeMessageOrderlyService.this.submitConsumeRequestLater(processQueue, mq, 3000);
				}
			}
		}, delayMills, TimeUnit.MILLISECONDS);
	}

	public ConsumerStatsManager getConsumerStatsManager() {
		return this.defaultMQPushConsumerImpl.getConsumerStatsManager();
	}

	class ConsumeRequest implements Runnable {
		private final ProcessQueue processQueue;
		private final MessageQueue messageQueue;

		public ConsumeRequest(ProcessQueue processQueue, MessageQueue messageQueue) {
			this.processQueue = processQueue;
			this.messageQueue = messageQueue;
		}

		@Override
		public void run() {
			// 1、检查ProcessQueue.dropped是否为true，若不是则直接返回；
			if (this.processQueue.isDropped()) {
				log.warn("run, the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
				return;
			}

			// 顺序消费三把锁
			// 锁2=====》在进行消息队列的消费过程中，对MessageQueue对象进行本地同步锁，保证同一时间只允许一个线程消息一个ConsumeQueue队列
			// **************************************
			// 2、对MessageQueue对象加锁，保证同一时间只允许一个线程使用该MessageQueue对象。
			// 调用ConsumeMessageOrderlyService.messageQueueLock. fetchLockObject(MessageQueue mq)获取锁对象。
			// 下面的处理逻辑均在获得该Object对象的互斥锁（synchronized）后进行处理；
			// 从而保证了在并发情况下一个MessageQueue对象只有一个线程使用
			final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
			synchronized (objLock) {
				// 3、若消息模式是广播
				// 或者对应的ProcessQueue.locked等于true
				// 且锁的时间未过期（根据获取锁locked的时候设置的lastLockTimestamp值来判断）,
				if (MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel()) || (this.processQueue.isLocked() && !this.processQueue.isLockExpired())) {
					final long beginTime = System.currentTimeMillis();

					// 则初始化局部变量continueConsume=true，然后无限期的循环执行下面的逻辑，直到局部变量continueConsume=false为止或者跳出循环，便终止了该方法的执行，
					// 否则执行第4步操作；
					// 3.1)执行for循环下面的逻辑，直到该continueConsume等于false或者直接跳出循环为止；
					// 3.11）继续从第3.1步开始遍历，不间断地从ProcessQueue对象的List<MessageExt>列表中获取消息对象并消费；直到消费完为止；
					for (boolean continueConsume = true; continueConsume;) {

						// 3.2）检查ProcessQueue.dropped是否为true，若不是则跳出循环；
						if (this.processQueue.isDropped()) {
							log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
							break;
						}

						// 3.3）若消息模式是集群并且ProcessQueue.locked不等于true（未锁住）或者锁住了但是锁已经超时
						if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel()) && !this.processQueue.isLocked()) {
							log.warn("the message queue not locked, so consume later, {}", this.messageQueue);
							// ***************************************************
							// 重新获取分布式锁后再消费

							// 则调用ConsumeMessageOrderlyService.tryLockLaterAndReconsume(MessageQueue mq, ProcessQueue processQueue, long delayMills)方法，
							// 在该方法中初始化一个线程并将线程放入ConsumeMessageOrderlyService. scheduledExecutorService线程池中，然后跳出循环。
							// 该线程在延迟delayMills即10毫秒之后被执行，
							// 该线程的功能是获取MessageQueue队列的分布式锁，
							// 然后再调用ConsumeMessageOrderlyService.submitConsumeRequestLater (MessageQueue mq, ProcessQueue processQueue, long delayMills)方法。
							ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
							break;
						}

						// 同上3.3
						if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel()) && this.processQueue.isLockExpired()) {
							log.warn("the message queue lock expired, so consume later, {}", this.messageQueue);
							// ***************************************************
							// 重新获取分布式锁后再消费
							ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
							break;
						}

						long interval = System.currentTimeMillis() - beginTime;
						// 3.4）若该循环从开始到现在连续执行的时间已经超过的最大连续执行值MaxTimeConsumeContinuously
						// （由property属性中的"rocketmq.client.maxTimeConsumeContinuously"设定，默认为60秒）
						if (interval > MaxTimeConsumeContinuously) {
							// 调用ConsumeMessageOrderlyService. submitConsumeRequestLater(MessageQueue mq, ProcessQueue processQueue, long delayMills)方法，
							// 其中delayMills=100，表示在100毫秒之后再调用ConsumeMessageOrderlyService.submitConsumeRequest(List<MessageExt> msgs, ProcessQueue processQueue, MessageQueue messageQueue, boolean dispathToConsume)方法，
							// 在该方法中重新创建ConsumeRequest对象，并放入ConsumeMessageOrderlyService.consumeExecutor线程池中重新被消费；
							ConsumeMessageOrderlyService.this.submitConsumeRequestLater(processQueue, messageQueue, 10);
							break;
						}

						// 3.4）获取一次批量消费的消息个数batchSize
						// （由参数DefaultMQPushConsumer.consumeMessageBatchMaxSize指定，默认为1）
						final int consumeBatchSize = ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();

						// 从ProcessQueue.msgTreeMap变量中获取batchSize个数的List<MessageExt>列表；并且从msgTreeMap中删除，存入临时变量msgTreeMapTemp中，返回List<MessageExt>列表，
						// 若该列表为空，表示该msgTreeMap列表中的消息已经消费完了，置ProcessQueue.consuming等于false；
						List<MessageExt> msgs = this.processQueue.takeMessags(consumeBatchSize);

						// 3.5）检查返回的List<MessageExt>列表是否为空
						if (!msgs.isEmpty()) {
							final ConsumeOrderlyContext context = new ConsumeOrderlyContext(this.messageQueue);
							ConsumeOrderlyStatus status = null;
							ConsumeMessageContext consumeMessageContext = null;

							// 3.6）检查DefaultMQPushConsumerImpl.consumeMessageHookList: ArrayList<ConsumeMessageHook>是否为空
							if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
								// 若不是，则初始化ConsumeMessageContext对象，
								consumeMessageContext = new ConsumeMessageContext();
								consumeMessageContext.setConsumerGroup(ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumerGroup());
								consumeMessageContext.setMq(messageQueue);
								consumeMessageContext.setMsgList(msgs);
								consumeMessageContext.setSuccess(false);

								// 并调用ArrayList<ConsumeMessageHook>列表中每个ConsumeMessageHook对象的consumeMessageBefore (ConsumeMessageContextcontext)方法，
								// 该ArrayList<ConsumeMessageHook>列表由业务层调用 DefaultMQPushConsumerImpl.registerConsumeMessageHook (ConsumeMessageHook hook)方法设置；
								ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.executeHookBefore(consumeMessageContext);
							}

							long beginTimestamp = System.currentTimeMillis();

							try {
								// 顺序消费锁
								// 锁3=====》在回调业务层定义的ConsumeMessageOrderlyService.messageListener:MessageListenerOrderly类的consumeMessage方法之前获取ProcessQueue.lockConsume:ReentrantLock变量的锁即消费处理队列的锁，该锁的粒度比消息队列的同步锁粒度更小，该锁的目的是保证在消费的过程中不会被解锁。
								this.processQueue.getLockConsume().lock();
								if (this.processQueue.isDropped()) {
									log.warn("consumeMessage, the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
									break;
								}

								// 3.7）执行业务层定义的消费消息的业务逻辑并返回消费结果
								/**
								 * 先调用ProcessQueue. lockConsume:ReentrantLock变量的lock方法获取锁（目的是防止在消费的过程中，被其他线程将此消费队列解锁了，从而引起并发消费的问题），
								 * 然后调用ConsumeMessageOrderlyService. messageListener:MessageListenerOrderly类的consumeMessage方法，
								 * 保证同一个ProcessQueue在同一时间只能有一个线程调用consumeMessage方法，
								 * 由应用层实现该MessageListenerOrderly接口的consumeMessage方法，
								 * 执行完成之后调用ProcessQueue. lockConsume:ReentrantLock变量的unlock方法释放锁；
								 */
								status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);
							} catch (Throwable e) {
								log.warn("consumeMessage exception: {} Group: {} Msgs: {} MQ: {}", //
										RemotingHelper.exceptionSimpleDesc(e), //
										ConsumeMessageOrderlyService.this.consumerGroup, //
										msgs, //
										messageQueue);
							} finally {
								this.processQueue.getLockConsume().unlock();
							}

							if (null == status //
									|| ConsumeOrderlyStatus.ROLLBACK == status//
									|| ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT == status) {
								log.warn("consumeMessage Orderly return not OK, Group: {} Msgs: {} MQ: {}", //
										ConsumeMessageOrderlyService.this.consumerGroup, //
										msgs, //
										messageQueue);
							}

							long consumeRT = System.currentTimeMillis() - beginTimestamp;

							if (null == status) {
								// 3.8）当consumeMessage方法的返回值status为空时，
								// 将结果状态status赋值为ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT；
								status = ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
							}

							// 3.9）检查DefaultMQPushConsumerImpl.consumeMessageHookList: ArrayList<ConsumeMessageHook>是否为空
							if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
								// 若不是，则初始化ConsumeMessageContext对象，
								consumeMessageContext.setStatus(status.toString());
								consumeMessageContext.setSuccess(ConsumeOrderlyStatus.SUCCESS == status || ConsumeOrderlyStatus.COMMIT == status);

								// 并调用ArrayList<ConsumeMessageHook>列表中每个ConsumeMessageHook对象的consumeMessageAfter (ConsumeMessageContextcontext)方法，
								// 该ArrayList<ConsumeMessageHook>列表由业务层调用DefaultMQPushConsumerImpl.registerConsumeMessageHook(ConsumeMessageHook hook)方法设置
								ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.executeHookAfter(consumeMessageContext);
							}

							ConsumeMessageOrderlyService.this.getConsumerStatsManager().incConsumeRT(ConsumeMessageOrderlyService.this.consumerGroup, messageQueue.getTopic(), consumeRT);

							// 3.10）处理回调方法consumeMessage的消费结果，并将消费结果赋值给变量continueConsume。
							// 调用ConsumeMessageOrderlyService. processConsumeResult(List<MessageExt> msgs, ConsumeOrderlyStatus status, ConsumeOrderlyContext context, ConsumeRequest consumeRequest)方法，
							// *****************************************
							// 根据消费结果进行相应处理
							continueConsume = ConsumeMessageOrderlyService.this.processConsumeResult(msgs, status, context, this);
						} else {
							// 若返回的List<MessageExt>列表为空，则置局部变量continueConsume=false
							// 表示不在循环执行，本次消息队列的消费结束；
							continueConsume = false;
						}
					}
				} else {
					if (this.processQueue.isDropped()) {
						log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
						return;
					}

					// ***************************************************
					// 重新获取分布式锁后再消费
					// 在集群模式下，若ProcessQueue未锁或者锁已经超时，
					// 则调用ConsumeMessageOrderlyService.tryLockLaterAndReconsume(MessageQueue mq, ProcessQueue processQueue, long delayMills)方法
					// 从Broker重新获取锁之后再进行消费。
					ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 100);
				}
			}
		}

		public ProcessQueue getProcessQueue() {
			return processQueue;
		}

		public MessageQueue getMessageQueue() {
			return messageQueue;
		}
	}

	public boolean processConsumeResult(//
			final List<MessageExt> msgs, //
			final ConsumeOrderlyStatus status, //
			final ConsumeOrderlyContext context, //
			final ConsumeRequest consumeRequest//
	) {
		boolean continueConsume = true;
		long commitOffset = -1L;
		if (context.isAutoCommit()) {
			switch (status) {
				case COMMIT :
				case ROLLBACK :
					log.warn("the message queue consume result is illegal, we think you want to ack these message {}", consumeRequest.getMessageQueue());

					// 1.1）若status等于SUCCESS，
				case SUCCESS :
					// ************************
					// 则调用ProcessQueue.commit()方法A--D
					// E)返回最大的offset+1的值并赋值给局部变量commitOffset值用于更新消费进度之用；
					commitOffset = consumeRequest.getProcessQueue().commit();
					// F）然后调用ConsumerStatsManager.incConsumeOKTPS(String group, String topic, long msgs)方法进行消费统计；
					this.getConsumerStatsManager().incConsumeOKTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
					break;

				// 1.2）若status等于SUSPEND_CURRENT_QUEUE_A_MOMENT，稍后重新消费。
				case SUSPEND_CURRENT_QUEUE_A_MOMENT :
					// 1.2.1）首先调用ProcessQueue.makeMessageToCosumeAgain (List<MessageExt> msgs)方法
					// 将List<MessageExt>列表的对象重新放入msgTreeMap变量中
					consumeRequest.getProcessQueue().makeMessageToCosumeAgain(msgs);

					// 1.2.2）然后调用ConsumeMessageOrderlyService. submitConsumeRequestLater(ProcessQueue processQueue, MessageQueue messageQueue, long suspendTimeMillis)方法，
					// 其中suspendTimeMillis由ConsumeOrderlyContext.suspendCurrentQueueTimeMillis变量在应用层的回调方法中设定，默认为1000，
					// 表示在1000毫秒之后调用ConsumeMessageOrderlyService.submitConsumeRequest(List<MessageExt> msgs, ProcessQueue processQueue, MessageQueue messageQueue, boolean dispathToConsume)方法，
					// 在该方法中重新创建ConsumeRequest对象，并放入ConsumeMessageOrderlyService.consumeExecutor线程池中；
					this.submitConsumeRequestLater(//
							consumeRequest.getProcessQueue(), //
							consumeRequest.getMessageQueue(), //
							context.getSuspendCurrentQueueTimeMillis());

					// 1.2.3）置局部变量continueConsume=false；
					continueConsume = false;

					// 1.2.4）调用ConsumerStatsManager.IncConsumeFailedTPS方法进行消费统计；
					this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
					break;
				default :
					break;
			}
		} else {
			// 2、若ConsumeOrderlyContext.autoCommit为false，根据consumeMessage方法返回的不同状态执行不同的逻辑：
			switch (status) {
				case SUCCESS :
					// 2.1）若status等于SUCCESS，仅调用ConsumerStatsManager.incConsumeOKTPS(String group, String topic, long msgs)方法进行消费统计，
					// 并未调用ProcessQueue.commit()方法清理队列数据；
					this.getConsumerStatsManager().incConsumeOKTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
					break;
				case COMMIT :
					// 2.2）若status等于COMMIT时，才调用ProcessQueue.commit()方法，并返回消费最大的offset值并赋值给局部变量commitOffset值用于更新消费进度之用；
					commitOffset = consumeRequest.getProcessQueue().commit();
					break;

				// 2.3）若status等于ROLLBACK，表示要重新消费，
				case ROLLBACK :
					// A）首先调用ProcessQueue.rollback()方法，将msgTreeMapTemp变量中的内容全部重新放入msgTreeMap变量中，同时清理msgTreeMapTemp变量；
					consumeRequest.getProcessQueue().rollback();

					// B）然后调用ConsumeMessageOrderlyService.submitConsumeRequestLater (ProcessQueue processQueue, MessageQueue messageQueue, long suspendTimeMillis)方法，
					// 其中suspendTimeMillis由ConsumeOrderlyContext. suspendCurrentQueueTimeMillis变量在应用层的回调方法中设定，默认为1000，表示在1000毫秒之后调用ConsumeMessageOrderlyService. submitConsumeRequest(List<MessageExt> msgs, ProcessQueue processQueue, MessageQueue messageQueue, boolean dispathToConsume)方法，
					// 在该方法中重新创建ConsumeRequest对象，并放入ConsumeMessageOrderlyService. consumeExecutor线程池中；
					this.submitConsumeRequestLater(//
							consumeRequest.getProcessQueue(), //
							consumeRequest.getMessageQueue(), //
							context.getSuspendCurrentQueueTimeMillis());

					// C）置局部变量continueConsume=false；
					continueConsume = false;
					break;

				// 2.4）若status等于SUSPEND_CURRENT_QUEUE_A_MOMENT，
				// 将msgTreeMapTemp变量中的内容全部重新放入msgTreeMap变量中，然后结束此轮消费（continueConsume=false），等待1秒之后再次提交消费，
				case SUSPEND_CURRENT_QUEUE_A_MOMENT :
					// A）首先调用ProcessQueue.makeMessageToCosumeAgain(List<MessageExt> msgs)方法
					consumeRequest.getProcessQueue().makeMessageToCosumeAgain(msgs);

					// B）然后调用ConsumeMessageOrderlyService.submitConsumeRequestLater (ProcessQueue processQueue, MessageQueue messageQueue, long suspendTimeMillis)方法，
					// 其中suspendTimeMillis由ConsumeOrderlyContext. suspendCurrentQueueTimeMillis变量在应用层的回调方法中设定，默认为1000
					this.submitConsumeRequestLater(//
							consumeRequest.getProcessQueue(), //
							consumeRequest.getMessageQueue(), //
							context.getSuspendCurrentQueueTimeMillis());

					// C）置局部变量continueConsume=false；
					continueConsume = false;

					// D）调用ConsumerStatsManager.IncConsumeFailedTPS方法进行消费统计；
					this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
					break;
				default :
					break;
			}
		}

		// 3、若上面两步得到的commitOffset值大于0
		if (commitOffset >= 0) {
			// 调用OffsetStore.updateOffset(MessageQueue mq, long offset, boolean increaseOnly)方法更新消费进度。
			// 对于LocalFileOffsetStore或RemoteBrokerOffsetStore类，该方法逻辑是一样的
			/**
			 * 以MessageQueue对象为key值从offsetTable: ConcurrentHashMap<MessageQueue, AtomicLong>变量中获取values值，
			 * 若该values值为空，则将MessageQueue对象以及commitOffset值存入offsetTable变量中，
			 * 若不为空，则比较已经存在的值，若大于已存在的值才更新；
			 */
			this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(consumeRequest.getMessageQueue(), commitOffset, false);
		}

		// 4、返回continueConsume值；若该值为false则结束此轮消费。
		return continueConsume;
	}

	private void submitConsumeRequestLater(//
			final ProcessQueue processQueue, //
			final MessageQueue messageQueue, //
			final long suspendTimeMillis//
	) {
		long timeMillis = suspendTimeMillis;
		if (timeMillis < 10) {
			timeMillis = 10;
		} else if (timeMillis > 30000) {
			timeMillis = 30000;
		}

		this.scheduledExecutorService.schedule(new Runnable() {

			@Override
			public void run() {
				ConsumeMessageOrderlyService.this.submitConsumeRequest(null, processQueue, messageQueue, true);
			}
		}, timeMillis, TimeUnit.MILLISECONDS);
	}

	@Override
	public void submitConsumeRequest(//
			final List<MessageExt> msgs, //
			final ProcessQueue processQueue, //
			final MessageQueue messageQueue, //
			final boolean dispathToConsume) {
		if (dispathToConsume) {
			// 若dispathToConsume=true（即上一次已经消费完了）
			// 则以ProcessQueue和MessageQueue对象为参数初始化ConsumeMessageOrderlyService类的内部线程类ConsumeRequest；
			// 然后将该线程类放入ConsumeMessageOrderlyService.consumeExecutor线程池中。
			// 从而可以看出顺序消费是从ProcessQueue对象的TreeMap树形列表中取消息的。
			ConsumeRequest consumeRequest = new ConsumeRequest(processQueue, messageQueue);
			this.consumeExecutor.submit(consumeRequest);
		}
		// 若上次的msgTreeMap变量中的数据还未消费完（即在2.4步中返回dispathToConsume=false）则不执行任何逻辑
	}

	@Override
	public void updateCorePoolSize(int corePoolSize) {
		if (corePoolSize > 0 //
				&& corePoolSize <= Short.MAX_VALUE //
				&& corePoolSize < this.defaultMQPushConsumer.getConsumeThreadMax()) {
			this.consumeExecutor.setCorePoolSize(corePoolSize);
		}
	}

	@Override
	public void incCorePoolSize() {
	}

	@Override
	public void decCorePoolSize() {
	}

	@Override
	public int getCorePoolSize() {
		return this.consumeExecutor.getCorePoolSize();
	}

	@Override
	public ConsumeMessageDirectlyResult consumeMessageDirectly(MessageExt msg, String brokerName) {
		ConsumeMessageDirectlyResult result = new ConsumeMessageDirectlyResult();
		result.setOrder(true);

		List<MessageExt> msgs = new ArrayList<MessageExt>();
		msgs.add(msg);
		MessageQueue mq = new MessageQueue();
		mq.setBrokerName(brokerName);
		mq.setTopic(msg.getTopic());
		mq.setQueueId(msg.getQueueId());

		ConsumeOrderlyContext context = new ConsumeOrderlyContext(mq);

		final long beginTime = System.currentTimeMillis();

		log.info("consumeMessageDirectly receive new messge: {}", msg);

		try {
			ConsumeOrderlyStatus status = this.messageListener.consumeMessage(msgs, context);
			if (status != null) {
				switch (status) {
					case COMMIT :
						result.setConsumeResult(CMResult.CR_COMMIT);
						break;
					case ROLLBACK :
						result.setConsumeResult(CMResult.CR_ROLLBACK);
						break;
					case SUCCESS :
						result.setConsumeResult(CMResult.CR_SUCCESS);
						break;
					case SUSPEND_CURRENT_QUEUE_A_MOMENT :
						result.setConsumeResult(CMResult.CR_LATER);
						break;
					default :
						break;
				}
			} else {
				result.setConsumeResult(CMResult.CR_RETURN_NULL);
			}
		} catch (Throwable e) {
			result.setConsumeResult(CMResult.CR_THROW_EXCEPTION);
			result.setRemark(RemotingHelper.exceptionSimpleDesc(e));

			log.warn(String.format("consumeMessageDirectly exception: %s Group: %s Msgs: %s MQ: %s", //
					RemotingHelper.exceptionSimpleDesc(e), //
					ConsumeMessageOrderlyService.this.consumerGroup, //
					msgs, //
					mq), e);
		}

		result.setAutoCommit(context.isAutoCommit());
		result.setSpentTimeMills(System.currentTimeMillis() - beginTime);

		log.info("consumeMessageDirectly Result: {}", result);

		return result;
	}

}
